package com.gitee.ywj1352.kafka;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;
import java.util.Random;

/**
 * @author wenjie.yang on 2019/6/21.
 * 本类的作用是分区算法
 * 默认的算发 也是 hash %
 */
public class MyPartition implements Partitioner {
    private Random random = new Random();

    /**
     * 分区算法 很关键 一定分配均匀
     * @param topic
     * @param key
     * @param keyBytes
     * @param value
     * @param valueBytes
     * @param cluster
     * @return
     */
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        //获取分区列表
        List<PartitionInfo> partitionInfos = cluster.partitionsForTopic(topic);
        int partitionNum = 0;
        if (key== null){
            partitionNum =  random.nextInt(partitionInfos.size());
        }else{
            partitionNum = Math.abs(key.hashCode() % partitionInfos.size());
        }
        System.out.println("key: " +key+";value :"+ value +"| "+ partitionNum );
        return partitionNum; //制定发送的分区值
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
